package com.github.kezhenxu94.playground.java.kafka;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class DemoConsumer {

	Properties properties;
	KafkaConsumer<String, String> consumer;
	volatile boolean stop = false;

	public DemoConsumer() throws InterruptedException {
		properties = new Properties();
		properties.setProperty("bootstrap.servers", "localhost:9092");
		properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
		properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		properties.setProperty("group.id", "GroupA");
		consumer = new KafkaConsumer<>(properties);
		consumer.subscribe(Arrays.asList("test"));
		ConsumerRecords<String, String> poll;
		while (!stop) {
			poll = consumer.poll(0);
			for (ConsumerRecord<String, String> r : poll) {
				System.out.println(r.offset() + " " + r.key() + " " + r.value());
				stop = "bye".equalsIgnoreCase(r.value());
			}
			Thread.sleep(100);
		}
		consumer.close();
	}

	public static void main(String[] args) throws InterruptedException {
		new DemoConsumer();
	}
}
